package pri.zxx.learndemo.threademo.futureTaskDemo;

import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import pri.zxx.learndemo.lambdaandstream.functions.Out;

import java.util.Optional;
import java.util.concurrent.*;
import java.util.stream.Stream;

/**
 * @author Administrator
 * @desc
 * @createTime 2019-04-01-下午 1:25
 */
@Slf4j
public class FutureTaskTest {
    //建立线程池
    private static ThreadPoolExecutor executor;

    static {
        executor = new ThreadPoolExecutor(
                1,
                20,
                20L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(), new DefaultThreadFactory("这是线程池名哦"));
    }


    @Test
    public void futureTest() throws ExecutionException, InterruptedException, TimeoutException {
        Future<Integer> result = getIntegerFuture();
        //获取结果
        Integer sum = result.get(2, TimeUnit.SECONDS);
        log.warn("sum:{}", sum);

        FutureTask ft = createTsk(2);
        executor.submit(ft);
        log.warn("total:{}", ft.get());

        executor.submit(ft);
        if (ft.isDone()) {
            log.warn("task done? {}", ft.isDone());
            log.warn("task result: {}", ft.get(1, TimeUnit.SECONDS));
        } else {
            log.error("任务出错");
        }
    }

    /**
     * 创建tsk
     */
    private Future<Integer> getIntegerFuture() {
        //执行任务
        return executor.submit(() -> {
            Optional<Integer> reduce = Stream.iterate(1, integer -> integer + 1).limit(500).reduce(Integer::sum);
            return reduce.orElse(0);
        });
    }

    /**
     * 创建tsk
     */
    private FutureTask createTsk(int start) {
        return new FutureTask(() -> {
            Optional<Integer> reduce = Stream.iterate(1, s -> s + start).peek(Out::print).limit(10).reduce(Integer::sum);
            return reduce.orElse(0);
        });
    }
}
